#!/usr/bin/env python3
# coding: utf-8
"""
acequant.marketdata.stock.tick  全市场
acequant.marketdata.stock.tick.xx 根据市场 (xshe)
acequant.marketdata.stock.tick.xx.xx 单标的 ()
"""
import asyncio
import json
import logging
import queue
import signal
import threading
from datetime import datetime, time, timedelta

import zmq
from google.protobuf.json_format import MessageToDict
from zmq import Context

import redis
from qp_base.const import TaskSignal
from qp_base.log import setup_logger
from qp_base.options import get_option as get_common_option
from qp_base.pbproto import quote_pb2
from qp_dq.basic import get_stock_trade_cal

from qp_dlive.options import get_option


class CNStockQuoteClient:
    """the china stock quote client"""

    quant_start_time = time(9, 10)
    quant_stop_time = time(15, 10)

    def __init__(self):
        # self.url = get_option("quotes_url")
        self.url = "tcp://localhost:9999"
        self.ctx = Context.instance()
        redis_url = get_common_option("qoute_redis_server", "redis://localhost:6379")
        self.redis_client = redis.from_url("redis://127.0.0.1:6379")
        self._task_queue: queue.Queue = queue.Queue()
        self._quit_event = threading.Event()

        # log config
        self.logger = logging.getLogger(__name__)
        setup_logger(self.logger, "qp_dlive.log")
        self.logger.info("qp_dlive started, quote server: %s", self.url)
        # Set the signal handler and a sigint
        signal.signal(TaskSignal.STOP, self.sig_handler)

        self._tick_thread = threading.Thread(target=self.tick_worker)
        self._tick_thread.start()

    def save_tick(self, msg, qlen: int):
        recv_dt = datetime.now()
        tick_msg = quote_pb2.StockTick()
        tick_msg.ParseFromString(msg)
        json_msg = MessageToDict(
            tick_msg, including_default_value_fields=True, use_integers_for_enums=True
        )
        symbol = ".".join([json_msg["symbol"], json_msg["market"]])
        tick_data = {
            "symbol": ".".join([json_msg["symbol"], json_msg["market"]]),
            "current": json_msg["lastPrice"],
            "pre_close": json_msg["preClosePrice"],
            "pre_settle": json_msg["preSettlementPrice"],
            "day_open": json_msg["openPrice"],
            "day_high": json_msg["highPrice"],
            "day_low": json_msg["lowPrice"],
            "day_avg": json_msg["averagePrice"],
            "day_volume": json_msg["totalVolume"],
            "day_amount": json_msg["totalValue"],
            "day_close": json_msg["closePrice"],
            "day_settle": json_msg["settlementPrice"],
            "limit_up": json_msg["UpperLimitPrice"],
            "limit_down": json_msg["LowerLimitPrice"],
            "recv_ts": int(json_msg["timeStamp"]),
            "phaseCode": "E0",
            "tick_volume": json_msg["volume"],
            "pe_ratio1": json_msg["peRatio1"],
            "pe_ratio2": json_msg["peRatio2"],
            "preCloseIOPV": 0.0,
            "iOPV": 0.0,
            "dt": recv_dt.timestamp(),
        }
        for ind, bid in enumerate(json_msg["bidPriceList"]):
            tick_data[f"bid{ind}"] = bid
        for ind, vol in enumerate(json_msg["bidVolumeList"]):
            tick_data[f"bid_vol{ind}"] = vol
        for ind, ask_price in enumerate(json_msg["askPriceList"]):
            tick_data[f"ask{ind}"] = ask_price
        for ind, vol in enumerate(json_msg["askVolumeList"]):
            tick_data[f"ask_vol{ind}"] = vol
        return self.redis_client.rpush(f"tick_quotes.{symbol}", json.dumps(tick_data))

    def subscriber_tick_data(self):
        # setup subscriber
        sub = self.ctx.socket(zmq.SUB)
        sub.connect(self.url)
        sub.setsockopt(zmq.SUBSCRIBE, b"acequant.marketdata.stock.tick")

        '------------关闭时间限制--------'
        # while (
        #     datetime.now().time() > self.quant_start_time
        #     and datetime.now().time() <= self.quant_stop_time
        # ):
        while True:
            try:
                [topic, msg] = sub.recv_multipart()
                # 其他时间段的数据都不要
                if msg:
                    self._task_queue.put(msg)
            except Exception as e:
                logging.exception("receive tick quotes exception:")
                # reconnect
                # sub.close()
                # sub = self.ctx.socket(zmq.SUB)
                # sub.connect(self.url)
                # sub.setsockopt(zmq.SUBSCRIBE, b"acequant.marketdata.stock.tick")
        #sub.close()

    def tick_worker(self):
        start_log_time = 0
        while True:
            try:
                msg = self._task_queue.get()
                self._task_queue.task_done()
                if not msg:
                    break
                if datetime.now().timestamp() - start_log_time > 600:
                    start_log_time = datetime.now().timestamp()
                    # print qlen log every ten minutes
                    self.logger.info(
                        "received quote data[qlen:%s]", self._task_queue.qsize()
                    )
                self.save_tick(msg, self._task_queue.qsize())
            except Exception:
                self.logger.exception("bg worker exception:")

    def sig_handler(self, signum, frame):
        self._task_queue.put(None)
        self._quit_event.set()

    def clear_redis_quotes(self):
        for key in self.redis_client.keys("tick_quotes.*"):
            return self.redis_client.delete(key)

    def start(self):
        self.logger.info("system start to subscribe stock tick data.")
        while True:
            self.logger.info("prepare to receive data")
            self.subscriber_tick_data()
            '------------关闭时间限制--------'
            # # trade cal 文件每年更新，因此需要在遍历完成后重新获取
            # trade_cals = get_stock_trade_cal(
            #     start_dt=datetime.now() - timedelta(days=1)
            # )
            # for trade_day in trade_cals["cal_date"]:
            #     current_dt = datetime.now()
            #     self.logger.info("new day loop start: %s", current_dt)
            #     if current_dt.date() > trade_day.date() or (
            #         current_dt.date() == trade_day.date()
            #         and current_dt.time() > self.quant_stop_time
            #     ):
            #         self.logger.info(
            #             "trade_day: %s, current_dt: %s, continue to next trade_day",
            #             trade_day,
            #             current_dt,
            #         )
            #         continue
            #     if current_dt.time() > self.quant_stop_time:
            #         self.logger.info(
            #             "sleep to next trade day start, curret_dt: %s, trade_day: %s",
            #             current_dt,
            #             trade_day,
            #         )
            #         ntrade_day_start = datetime.combine(trade_day.date(), time())
            #         time2sleep = (ntrade_day_start - current_dt).total_seconds()
            #         if self._quit_event.wait(timeout=time2sleep):
            #             self.logger.info("quit event is set, will exit directly")
            #             return
            #         current_dt = datetime.now()
            #     if current_dt.time() < self.quant_start_time:
            #         # clear redis
            #         self.logger.info(
            #             "curret_dt: %s start clear redis quotes.", current_dt
            #         )
            #         self.clear_redis_quotes()
            #
            #     next_sub_date = trade_day.replace(hour=9, minute=10, second=0)
            #     time2sleep = (next_sub_date - current_dt).total_seconds()
            #     if time2sleep > 0:
            #         self.logger.info(
            #             "sleep %s seconds to %s", time2sleep, next_sub_date
            #         )
            #         if self._quit_event.wait(timeout=time2sleep):
            #             self.logger.info("quit event is set, will exit directly")
            #             return
            #     self.logger.info("prepare to receive data")
            #     self.subscriber_tick_data()


def main():
    cli = CNStockQuoteClient()
    # test 清除redis 旧数据
    cli.clear_redis_quotes()
    cli.start()


if __name__ == "__main__":
    main()
